package server

import (
	"XcRPC/codec"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"reflect"
	"strings"
	"sync"
	"time"
)

const MagicNumber = 0x1e961f

type Option struct {
	MagicNumber    int
	CodecType      codec.Type
	ConnectTimeout time.Duration
	HandleTimeout  time.Duration
}

var DefaultOption = &Option{
	MagicNumber:    MagicNumber,
	CodecType:      codec.GobType,
	ConnectTimeout: 10 * time.Second,
}

type Server struct {
	serviceMap sync.Map
}

func NewServer() *Server {
	return &Server{}
}

var DefaultServer = NewServer()

func (server *Server) Accept(lis net.Listener) {
	for {
		conn, err := lis.Accept() //阻塞等待conn连接
		if err != nil {
			log.Printf("rpc server: accept error:", err)
			return
		}
		log.Printf("Client connected:", conn.RemoteAddr())
		go server.ServeConn(conn)
	}
}

func Accept(lis net.Listener) {
	DefaultServer.Accept(lis)
}

func (server *Server) ServeConn(conn io.ReadWriteCloser) {
	defer func() {}()
	var option Option
	if err := json.NewDecoder(conn).Decode(&option); err != nil { //阻塞等待，解码option
		log.Println("rpc server: options error: ", err)
		return
	}

	if option.MagicNumber != MagicNumber {
		log.Printf("rpc server: invalid magic number %x", option.MagicNumber)
		return
	}

	codecFunc := codec.NewCodecFuncMap[option.CodecType]
	if codecFunc == nil {
		log.Printf("rpc server: invalid codec type %s", option.CodecType)
		return
	}
	timeout := option.HandleTimeout
	server.serverCodec(codecFunc(conn), timeout)
}

var invalidRequest = struct{}{}

func (server *Server) serverCodec(cc codec.Codec, timeout time.Duration) {
	sendLock := new(sync.Mutex)
	wg := sync.WaitGroup{}
	for {
		req, err := server.readRequest(cc)
		if err != nil {
			if req == nil {
				break
			}
			req.h.Error = err.Error()
			server.sendResponse(cc, req.h, invalidRequest, sendLock)
			continue
		}
		wg.Add(1)
		go server.handleRequest(cc, req, sendLock, &wg, timeout)
	}
	wg.Wait()
	cc.Close()
}

type request struct {
	h            *codec.Header
	argv, replyv reflect.Value
	mType        *methodType
	svc          *service
}

func (server *Server) readRequest(cc codec.Codec) (*request, error) {
	header, err := server.readRequestHeader(cc)
	if err != nil {
		return nil, err
	}
	req := &request{h: header}

	req.svc, req.mType, err = server.findService(header.ServiceMethod)
	if err != nil {
		return req, err
	}
	req.argv = req.mType.newArgv()
	req.replyv = req.mType.newReplyv()

	argvi := req.argv.Interface()
	if req.argv.Type().Kind() != reflect.Ptr {
		argvi = req.argv.Addr().Interface()
	}
	if err = cc.ReadBody(argvi); err != nil {
		log.Println("rpc server: read argv err:", err)
		return req, err
	}
	return req, nil
}

func (server *Server) readRequestHeader(cc codec.Codec) (*codec.Header, error) {
	var header codec.Header
	if err := cc.ReadHeader(&header); err != nil {
		log.Printf("rpc server: read header error: %v", err)
		if err == io.EOF || err == io.ErrUnexpectedEOF {
			return nil, err
		}

	}
	return &header, nil
}

func (server *Server) ReadRequestBody(cc codec.Codec) interface{} {
	var body interface{}
	if err := cc.ReadBody(&body); err != nil {
		log.Printf("rpc server: read body error: %v", err)
	}
	return body
}

func (server *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}, sendLock *sync.Mutex) {
	sendLock.Lock()
	defer sendLock.Unlock()
	if err := cc.Write(h, body); err != nil {
		log.Printf("rpc server: write response error: %v", err)
	}
}

func (server *Server) handleRequest(cc codec.Codec, req *request, sendLock *sync.Mutex, wg *sync.WaitGroup, timeout time.Duration) {
	defer wg.Done()

	done := make(chan struct{})
	var once sync.Once
	go func() {
		err := req.svc.call(req.mType, req.argv, req.replyv)
		if err != nil {
			req.h.Error = err.Error()
			once.Do(func() {
				server.sendResponse(cc, req.h, invalidRequest, sendLock)
				done <- struct{}{}
			})
			return
		}
	}()

	if timeout == 0 {
		<-done
		return
	}

	select {
	case <-time.After(timeout):
		once.Do(func() {
			req.h.Error = fmt.Sprintf("rpc server: request handle timeout: expect within %s", timeout)
			server.sendResponse(cc, req.h, invalidRequest, sendLock)
		})
		return
	case <-done:
		return
	}
}

func (server *Server) Register(rcvr interface{}) error {
	service := NewService(rcvr)
	if _, ok := server.serviceMap.LoadOrStore(service.name, service); ok {
		return errors.New("rpc: service already defined: " + service.name)
	}
	return nil
}

func Register(rcvr interface{}) error {
	return DefaultServer.Register(rcvr)
}

func (server *Server) findService(serviceMethod string) (*service, *methodType, error) {
	splitServiceMethod := strings.Split(serviceMethod, ".")
	if len(splitServiceMethod) != 2 {
		return nil, nil, errors.New("rpc server: service/method request ill-formed: " + serviceMethod)
	}
	serviceName, methodName := splitServiceMethod[0], splitServiceMethod[1]
	seri, ok := server.serviceMap.Load(serviceName)
	if !ok {
		return nil, nil, errors.New("rpc server: can't find service " + serviceName)
	}
	service := seri.(*service)
	method := service.methods[methodName]
	if method == nil {
		return nil, nil, errors.New("rpc server: can't find method " + methodName)
	}
	return service, method, nil
}
